"""
Case Type   : 备机支持逻辑复制
Case Name   : 备机执行switchover后，备机执行逻辑复制
Create At   : 2021/04/20
Owner       : zopentestcase005
Description :
        1.备机上执行switchover，将备机升为主机
        2.修改参数wal_level为logical;enable_slot_log为on
        3.主备机pg_hba.conf文件中配置逻辑复制的用户白名单
        4.主机创建逻辑复制槽
        5.主机上查询逻辑复制槽
        6.备机执行逻辑解码
        7.主机建表并执行DML操作
        8.查看指定解码文件
        9.停止解码
        10.清理环境
Expect      :
        1.备机升为主机成功
        2.修改参数wal_level为logical;enable_slot_log为on成功
        3.pg_hba.conf 配置逻辑复制的用户白名单成功
        4.主机创建逻辑复制槽成功
        5.显示slot_test022复制槽信息
        6.解码命令执行成功，显示解码过程
        7.主机建表并执行DML操作成功
        8.解码文件解析DML操作成功
        9.成功
        10.清理环境完成
History     :modified by zopentestcase005,2021/10/12,修改用例，适配2.1.0版本
             modified by zopentestcase005,2022.05.07，修改用例中查看解码文件步骤
             modified by zopentestcase005,2022.05.30，增加等待主备同步防止备机获取解码文件时
             未同步问题
"""
import os
import time
import unittest

from testcase.utils.ComThread import ComThread
from testcase.utils.Common import Common
from testcase.utils.CommonSH import CommonSH
from testcase.utils.Constant import Constant
from testcase.utils.Logger import Logger
from yat.test import Node
from yat.test import macro

Primary_SH = CommonSH('PrimaryDbUser')
constant = Constant()


@unittest.skipIf(1 == Primary_SH.get_node_num(),
                 '单机环境不执行')
class LogicalReplication(unittest.TestCase):
    def setUp(self):
        self.log = Logger()
        self.log.info(f'-----{os.path.basename(__file__)} start-----')
        self.Standby_SH = CommonSH('Standby1DbUser')
        self.com = Common()
        self.constant = Constant()
        self.primary_node = Node('PrimaryDbUser')
        self.standby_node = Node('Standby1DbUser')
        self.root_node = Node('PrimaryRoot')
        self.decode_file = os.path.join(macro.DB_INSTANCE_PATH,
                                        'logical22.txt')
        self.pg_hba = os.path.join(macro.DB_INSTANCE_PATH,
                                   macro.PG_HBA_FILE_NAME)
        self.tb_name = "replication0022"

    def test_standby(self):
        text = '--step1:执行switchover;expect:成功--'
        self.log.info(text)
        excute_cmd = f'''source {macro.DB_ENV_PATH};
                    gs_ctl switchover -D {macro.DB_INSTANCE_PATH} -m fast ;
                    '''
        self.log.info(excute_cmd)
        excute_msg = self.standby_node.sh(excute_cmd).result()
        self.log.info(excute_msg)
        self.assertIn(constant.SWITCHOVER_SUCCESS_MSG, excute_msg,
                      '执行失败' + text)
        self.log.info('--进行refreshconf--')
        excute_cmd = f'''source {macro.DB_ENV_PATH};
                    gs_om -t refreshconf;
                    '''
        self.log.info(excute_cmd)
        excute_msg = self.standby_node.sh(excute_cmd).result()
        self.log.info(excute_msg)
        self.assertIn(constant.REFRESHCONF_SUCCESS_MSG, excute_msg)
        self.log.info('---重启数据库--')
        restart_msg = Primary_SH.restart_db_cluster()
        self.log.info(restart_msg)
        self.log.info('---查看主备状态----')
        status_cmd = f'''source {macro.DB_ENV_PATH};
                    gs_om -t status --detail;
                    '''
        self.log.info(status_cmd)
        status_msg = self.primary_node.sh(status_cmd).result()
        self.log.info(status_msg)
        self.node_msg = status_msg.splitlines()[10].strip()
        self.log.info(self.node_msg)
        self.assertTrue("Degraded" in status_msg or "Normal" in status_msg,
                        '执行失败' + text)
        self.assertIn('Standby', self.node_msg, '执行失败' + text)

        text = '--step2:修改wal_level为logical;enable_slot_log为on;expect:成功--'
        self.log.info(text)
        mod_msg = self.Standby_SH.execute_gsguc('set',
                                                constant.GSGUC_SUCCESS_MSG,
                                                'wal_level =logical')
        self.log.info(mod_msg)
        self.assertTrue(mod_msg)
        mod_msg = self.Standby_SH.execute_gsguc('set',
                                                constant.GSGUC_SUCCESS_MSG,
                                                'enable_slot_log =on')
        self.log.info(mod_msg)
        self.assertTrue(mod_msg, '执行失败' + text)
        self.log.info('--重启数据库--')
        restart_msg = self.Standby_SH.restart_db_cluster()
        self.log.info(restart_msg)
        status = self.Standby_SH.get_db_cluster_status()
        self.assertTrue("Degraded" in status or "Normal" in status)

        text = '--step3:主备均配置逻辑复制的用户;expect:成功--'
        self.log.info(text)
        sql_cmd = self.Standby_SH.execut_db_sql(f'''drop role if exists rep;
            create role rep with login password '{macro.COMMON_PASSWD}';
            alter role rep with replication sysadmin;''')
        self.log.info(sql_cmd)
        self.assertIn(constant.CREATE_ROLE_SUCCESS_MSG, sql_cmd)
        self.assertIn(constant.ALTER_ROLE_SUCCESS_MSG, sql_cmd)
        mod_msg = f"sed -i '$a\local    replication     rep      trust' " \
                  f"{self.pg_hba}"
        self.log.info(mod_msg)
        msg = self.primary_node.sh(mod_msg).result()
        self.log.info(msg)
        mod_msg = f"sed -i '$a\host    replication     rep   127.0.0.1/32   " \
                  f"trust' {self.pg_hba}"
        self.log.info(mod_msg)
        msg = self.primary_node.sh(mod_msg).result()
        self.log.info(msg)
        mod_msg = f"sed -i '$a\host   replication     rep   ::1/128    " \
                  f"trust'  {self.pg_hba}"
        self.log.info(mod_msg)
        msg = self.primary_node.sh(mod_msg).result()
        self.log.info(msg)
        mod_msg = f"sed -i '$a\local    replication     rep      trust' " \
                  f"{self.pg_hba}"
        self.log.info(mod_msg)
        msg = self.standby_node.sh(mod_msg).result()
        self.log.info(msg)
        mod_msg = f"sed -i '$a\host    replication     rep   127.0.0.1/32   " \
                  f"trust' {self.pg_hba}"
        self.log.info(mod_msg)
        msg = self.standby_node.sh(mod_msg).result()
        self.log.info(msg)
        mod_msg = f"sed -i '$a\host    replication     rep   ::1/128  " \
                  f"trust' {self.pg_hba}"
        self.log.info(mod_msg)
        msg = self.standby_node.sh(mod_msg).result()
        self.log.info(msg)
        restart_msg = Primary_SH.restart_db_cluster()
        self.log.info(restart_msg)
        status = Primary_SH.get_db_cluster_status()
        self.assertTrue("Degraded" in status or "Normal" in status)

        text = '--step4:创建逻辑复制槽;expect:成功--'
        self.log.info(text)
        check_res = self.Standby_SH.execut_db_sql('select slot_name from '
                                                  'pg_replication_slots;')
        self.log.info(check_res)
        if 'slot_test022' in check_res.split('\n')[-2].strip():
            del_cmd = self.Standby_SH.execut_db_sql("select * from "
                                                    "pg_drop_replication_slot"
                                                    "('slot_test022');")
            self.log.info(del_cmd)
        cre_cmd = self.Standby_SH.execut_db_sql("select * from "
                                                "pg_create_logical_"
                                                "replication_slot"
                                                "('slot_test022', "
                                                "'mppdb_decoding');")
        self.log.info(cre_cmd)

        text = '--step5:查询复制槽;expect:成功--'
        self.log.info(text)
        query_cmd = self.Standby_SH.execut_db_sql(
            'select slot_name,plugin from'
            ' pg_get_replication_slots();')
        self.log.info(query_cmd)
        self.assertIn('slot_test022', query_cmd)
        query_cmd = Primary_SH.execut_db_sql(
            'select slot_name,plugin from'
            ' pg_get_replication_slots();')
        self.log.info(query_cmd)
        self.assertIn('slot_test022', query_cmd, '执行失败' + text)

        text = '--step6:备机执行逻辑复制槽流式解码;expect:显示解码过程--'
        self.log.info(text)
        decode_cmd = f"pg_recvlogical " \
                     f"-d {self.primary_node.db_name} " \
                     f"-S slot_test022 " \
                     f"-p {str(int(self.primary_node.db_port)+1)} " \
                     f"--start " \
                     f"-f {self.decode_file} " \
                     f"-s 2 " \
                     f"-v " \
                     f"-P mppdb_decoding " \
                     f"-U rep"
        execute_cmd = f'''source {macro.DB_ENV_PATH}
                          expect <<EOF
                          set timeout 300
                          spawn {decode_cmd}
                          expect "Password:"
                          send "{macro.COMMON_PASSWD}\\n"
                          expect eof\n''' + '''EOF'''
        self.log.info(execute_cmd)
        thread_2 = ComThread(self.com.get_sh_result,
                             args=(self.primary_node, execute_cmd))
        thread_2.setDaemon(True)
        thread_2.start()
        thread_2.join(30)

        text = '--step7:创建表并进行DML操作;expect:成功--'
        self.log.info(text)
        sql_cmd = self.Standby_SH.execut_db_sql(f'''drop table if exists  \
        {self.tb_name};
        create table {self.tb_name}(c_1 integer,c_2 bigint,c_3 smallint,\
        c_4 tinyint, c_5 serial,c_6 smallserial,c_7 bigserial,c_8 float,\
        c_9 double precision,c_10 date,c_11 time without time zone,\
        c_12 timestamp without time zone,c_13 char(10),c_14 varchar(20),\
        c_15 text,c_16 blob,c_17 bytea);
        insert into {self.tb_name} values(1,10,5,25,default,default,default,\
        1237.127,123456.1234,date '12-10-2010','21:21:21','2010-12-12',\
        '测试','测试工程师','西安',empty_blob(),E'\\xDEADBEEF');
        update {self.tb_name} set c_15 = '数据库';
        delete from {self.tb_name} where c_16 = empty_blob();''')
        self.log.info(sql_cmd)
        self.assertIn(constant.TABLE_CREATE_SUCCESS, sql_cmd, '执行失败' + text)

        time.sleep(25)
        self.log.info('等待主备一致')
        for i in range(90):
            result = Primary_SH.check_data_consistency()
            if result:
                break
            time.sleep(20)
            self.assertTrue(result)
        status = Primary_SH.get_db_cluster_status("detail")
        self.log.info(status)

        text = '--step8:备机查看解码文件;expect:成功--'
        self.log.info(text)
        cat_cmd = f"cat {self.decode_file};"
        self.log.info(cat_cmd)
        result = self.primary_node.sh(cat_cmd).result()
        self.log.info(result)
        du_cmd = f"du -h {self.decode_file};"
        self.log.info(du_cmd)
        du_msg = self.primary_node.sh(du_cmd).result()
        self.log.info(du_msg)
        dumsg_list = du_msg.split()[0]
        self.log.info(dumsg_list)
        self.assertTrue(float(dumsg_list[:-1]) > 0, '执行失败' + text)

        text = '--step9:停止解码;expect:成功--'
        self.log.info(text)
        stop_cmd = "ps -ef |  grep  pg_recvlogical | grep -v grep | " \
                   "awk '{{print $2}}' | xargs sudo kill -9"
        self.log.info(stop_cmd)
        result = self.root_node.sh(stop_cmd).result()
        self.log.info(result)

    def tearDown(self):
        text = '--step10:清理环境;expect:成功--'
        self.log.info(text)
        del_cmd = self.Standby_SH.execut_db_sql("select * from "
                                                "pg_drop_replication_slot"
                                                "('slot_test022');")
        self.log.info(del_cmd)
        self.assertIn('', del_cmd)
        drop_cmd = self.Standby_SH.execut_db_sql(f'''drop role if exists rep;
            drop table if exists {self.tb_name};''')
        self.log.info(drop_cmd)
        del_msg = f"sed -i '/replication     rep/d' {self.pg_hba}"
        self.log.info(del_msg)
        msg = self.primary_node.sh(del_msg).result()
        self.log.info(msg)
        del_msg = f"sed -i '/replication     rep/d' {self.pg_hba}"
        self.log.info(del_msg)
        msg = self.standby_node.sh(del_msg).result()
        self.log.info(msg)
        rm_cmd = f"rm -rf {self.decode_file};"
        result = self.primary_node.sh(rm_cmd).result()
        self.log.info(result)
        restore_cmd = self.Standby_SH.execute_gsguc('set',
                                                    constant.GSGUC_SUCCESS_MSG,
                                                    'wal_level=hot_standby')
        self.log.info(restore_cmd)
        restore_cmd = self.Standby_SH.execute_gsguc('set',
                                                    constant.GSGUC_SUCCESS_MSG,
                                                    'enable_slot_log=off')
        self.log.info(restore_cmd)
        restart_msg = self.Standby_SH.restart_db_cluster()
        self.log.info(restart_msg)
        status = Primary_SH.get_db_cluster_status()
        self.assertTrue("Degraded" in status or "Normal" in status)
        query_cmd = f'''source {macro.DB_ENV_PATH};
                   gs_om -t status --detail;
                   '''
        self.log.info(query_cmd)
        status_msg = self.primary_node.sh(query_cmd).result()
        self.log.info(status_msg)
        self.node_msg = status_msg.splitlines()[10].strip()
        self.log.info(self.node_msg)
        if 'Standby' in self.node_msg:
            self.log.info('--------------恢复主备状态--------------')
            recover_cmd = f'''source {macro.DB_ENV_PATH};
                             gs_ctl switchover -D {macro.DB_INSTANCE_PATH};
                             gs_om -t refreshconf;
                                    '''
            self.log.info(recover_cmd)
            recover_msg = self.primary_node.sh(recover_cmd).result()
            self.log.info(recover_msg)
        else:
            return '主备节点正常'
        self.assertIn(self.constant.DROP_ROLE_SUCCESS_MSG, drop_cmd,
                      '执行失败' + text)
        self.assertIn(self.constant.TABLE_DROP_SUCCESS, drop_cmd,
                      '执行失败' + text)
        self.log.info(f'-----{os.path.basename(__file__)} end-----')
